package com.atguigu.flink.datastreamapi.dirstribute;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/4
 *
 *  SlotSharingGroup: 就是一个标记，标记了当前的算子是否应该和其他的算子放在一个slot中运行。
 *
 *      默认，所有的算子的slotSharingGroup都是default.
 *      可以使用 算子.slotSharingGroup("xx") 指定当前算子的slot的标记。
 *      slotSharingGroup可以向下传递。
 *
 *      只有 slotSharingGroup相同的算子，才能放入一个slot中运行！
 *      不同slotSharingGroup的算子，是无法进行OperatorChain.
 *
 *  -----------------------
 *      slotSharingGroup的作用是把计算密集型和资源密集型的算子分开。
 *          举例:  a,b 都是计算密集型，消耗cpu
 *                 c,d 都是资源密集型，消耗内存
 *
 *                 a,c 在同一个 slotSharingGroup
 *                 b,d 在同一个 slotSharingGroup
 *
 *                 不能让 a,b 或 c,d 在同一个slotSharingGroup
 */
public class Demo8_SlotSharingGroup
{
    public static void main(String[] args) {

        //http://hadoop103:42218
        /*
                远程集群上没有的，本地的代码用到的jar包，才需要去提交。
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("hadoop103", 42218, "files/a.jar");


        env.setParallelism(2);

        env.socketTextStream("hadoop102", 8888).name("source")
           .map(x -> x).name("map1").slotSharingGroup("m1")
           .map(x -> x).name("map2").slotSharingGroup("m2")
           .map(x -> x).name("map3").slotSharingGroup("m3")
           .print().slotSharingGroup("p1");


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
